﻿
Imports System.Threading
Imports osi.root.delegates
Imports osi.root.constants
Imports osi.root.connector
Imports osi.root.envs
Imports osi.root.formation
Imports osi.root.lock

Namespace queue_runner
    Public Module _queue_runner
        Public ReadOnly thread_count As Int32
        Private ReadOnly LENGTH As Int64
        Private ReadOnly INTERVAL As Int64
        Private ReadOnly USED As Int64
        Private ReadOnly q As qless2(Of _do(Of Boolean))
        Private ReadOnly are As AutoResetEvent

        Sub New()
            thread_count = determine_thread_count()
            LENGTH = counter.register_average_and_last_average("QUEUE_RUNNER_LENGTH")
            INTERVAL = counter.register_average_and_last_average("QUEUE_RUNNER_INTERVAL_TICKS")
            USED = counter.register_average_and_last_average("QUEUE_RUNNER_USED_TICKS")
            q = New qless2(Of _do(Of Boolean))()
            If Not busy_wait Then
                are = New AutoResetEvent(False)
            End If
            start()
            application_lifetime.stopping_handle(Sub()
                                                     If Not are Is Nothing Then
                                                         are.Close()
                                                         GC.KeepAlive(are)
                                                     End If
                                                 End Sub)
        End Sub

        Private Function determine_thread_count() As Int32
            assert((queue_runner_thread_count > 0 AndAlso
                    queue_runner_thread_count <= Environment.ProcessorCount()) OrElse
                   queue_runner_thread_count = npos)
            Return If(queue_runner_thread_count <> npos,
                      queue_runner_thread_count,
                      max(1, Environment.ProcessorCount() >> 2))
        End Function

        Public Sub trigger()
            If Not are Is Nothing Then
                are.Set()
            End If
        End Sub

        Private Sub check_push(ByVal e As _do(Of Boolean))
            assert(Not e Is Nothing)
            If do_(e, False) Then
                q.emplace_push(e)
            End If
        End Sub

        Private Sub start()
            For i As Int32 = 0 To thread_count - 1
                Dim id As Int32 = 0
                id = i
                queue_in_managed_threadpool(Sub()
                                                assert(processor_affinity = npos OrElse
                                                       (processor_affinity >= 0 AndAlso
                                                        processor_affinity < Environment.ProcessorCount()))
                                                If processor_affinity <> npos Then
                                                    loop_set_thread_affinity(processor_affinity + id)
                                                End If
                                                While application_lifetime.running()
                                                    Dim size As Int64 = 0
                                                    size = q.size()
                                                    counter.increase(LENGTH, size)
                                                    If thread_count > 1 Then
                                                        size = Math.Ceiling(size / thread_count)
                                                    End If
                                                    counter.record_time_begin()
                                                    Dim e As _do(Of Boolean) = Nothing
                                                    While size > 0 AndAlso q.pop(e)
                                                        check_push(e)
                                                        size -= 1
                                                    End While
                                                    counter.record_time_ticks(USED)
                                                    counter.record_time_begin()
                                                    If Not are Is Nothing Then
                                                        yield_wait(are, queue_runner_interval_ms)
                                                    End If
                                                    counter.record_time_ticks(INTERVAL)
                                                End While
                                            End Sub)
            Next
        End Sub

        Public Function push(ByVal d As _do(Of Boolean)) As Boolean
            If d Is Nothing Then
                Return False
            Else
                check_push(d)
                Return True
            End If
        End Function

        Public Function size() As Int64
            Return q.size()
        End Function
    End Module
End Namespace
